/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hung.triple.heartbeat;

import com.hung.triple.consumer.response.TripleFuture;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * An abstraction of socket channel.
 *
 * @author yunliang.shi
 * @version $Id: Connection.java, v 0.1 Mar 10, 2016 11:30:54 AM yunliang.shi Exp $
 */
@Slf4j
public class Connection {
    public static final AttributeKey<Connection> CONNECTION = AttributeKey.valueOf("connection");
    public static final AttributeKey<Integer> HEARTBEAT_COUNT = AttributeKey.valueOf("heartbeatCount");
    public static final AttributeKey<Boolean> HEARTBEAT_SWITCH = AttributeKey.valueOf("heartbeatSwitch");
    private final ConcurrentHashMap<Integer, TripleFuture> invokeFutureMap;
    private Channel channel;
    private AtomicBoolean closed;

    public Connection(Channel channel) {
        this.invokeFutureMap = new ConcurrentHashMap(4);
        this.closed = new AtomicBoolean(false);
        this.channel = channel;
        this.channel.attr(CONNECTION).set(this);
    }

    public InetSocketAddress getRemoteAddress() {
        return (InetSocketAddress) this.channel.remoteAddress();
    }

    public Channel getChannel() {
        return this.channel;
    }

    public TripleFuture addInvokeFuture(TripleFuture future) {
        return this.invokeFutureMap.putIfAbsent(future.getInvokeId(), future);
    }

    public TripleFuture removeInvokeFuture(int id) {
        return this.invokeFutureMap.remove(id);
    }


    public void close() {
        if (this.closed.compareAndSet(false, true)) {
            try {
                if (this.getChannel() != null) {
                    this.getChannel().close().addListener(new ChannelFutureListener() {
                        public void operationComplete(ChannelFuture future) throws Exception {
                            Connection.log.info("Close the connection to remote address={}, result={}, cause={}", new Object[]{RemotingUtil.parseRemoteAddress(Connection.this.getChannel()), future.isSuccess(), future.cause()});
                        }
                    });
                }
            } catch (Exception var2) {
                log.warn("Exception caught when closing connection {}", RemotingUtil.parseRemoteAddress(this.getChannel()), var2);
            }
        }

    }


}
